package org.example;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Flink Event Time 和 Watermarks 流式分析示例
 * 本示例展示如何使用 Flink 的 Event Time 和 Watermarks 进行流式分析：
 * 1. 从文本流中读取数据，每条数据包含时间戳。
 * 2. 使用 WatermarkStrategy 设置 Event Time 和 Watermark 生成策略。
 * 3. 按单词分组，使用 TumblingEventTimeWindows 进行窗口聚合，统计每 5 秒内单词出现次数。
 * 适合初学者学习 Flink 时间语义和窗口计算的基本使用。
 */
public class EventTimeWatermarkExample {

    public static void main(String[] args) throws Exception {
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为 1，方便观察输出
        env.setParallelism(1);

        // 模拟输入数据流，每条数据包含时间戳（实际应用中可能来自 Kafka、Socket 等）
        DataStream<String> textStream = env.fromElements(
            "Hello Flink, 1625097600000", // 2021-07-01 00:00:00
            "Hello World, 1625097605000", // 2021-07-01 00:00:05
            "Hello Flink, 1625097610000"  // 2021-07-01 00:00:10
        );

        // 使用 FlatMap 将每行文本拆分为单词，并转换为 (word, 1) 的元组，同时提取时间戳
        DataStream<Tuple2<String, Integer>> wordCountStream = textStream
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
                    String[] parts = line.split(", ");
                    String text = parts[0];
                    long timestamp = Long.parseLong(parts[1]);
                    for (String word : text.split(" ")) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            })
            // 设置 Event Time 和 Watermark 生成策略
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Tuple2<String, Integer>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                    .withTimestampAssigner((event, timestamp) -> Long.parseLong(event.f0.split(", ")[1]))
            )
            // 按单词分组，使用 TumblingEventTimeWindows 进行窗口聚合，统计每 5 秒内单词出现次数
            .keyBy(0)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .sum(1);

        // 输出结果
        wordCountStream.print("Event Time 窗口结果");

        // 执行流处理作业
        env.execute("Event Time and Watermark Example");
    }
} 